package com.sunzm.flink.callrecordconnect

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 双流连接（connect），
 * 通话记录和满意度消息拼接需求
 * 使用状态编程来实现
 *
 * 存在的问题：
 * 如果只收到了一个消息，那么永远不会发出
 */
object CallRecordConnectJobV2 {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //获取通话记录消息
    val callLogDS: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //解析名keyBy
    val keyedCallLogDS: KeyedStream[String, String] = callLogDS.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {

        if (StringUtils.isNotBlank(value)) {

          try {
            //解析JSON
            val jSONObject = JSON.parseObject(value)

            val callRecordId = jSONObject.getString("callRecordId")
            if (StringUtils.isNotBlank(callRecordId)) {

              out.collect(value)

            } else {
              //打印告警信息
              logger.warn(s"缺少 callRecordId 字段: ${value}")
            }
          } catch {
            case e: Throwable => {
              logger.error(s"JSON解析异常: ${value}", e)
            }
          }

        }

      }
    }).keyBy(line => {
      val jSONObject = JSON.parseObject(line)

      jSONObject.getString("callRecordId")
    })

    //获取满意度消息
    val satisfactionDS: DataStream[String] = env.socketTextStream("82.156.210.70", 8888)

    val keyedSatisfactionDS: KeyedStream[String, String] = satisfactionDS.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {

        if (StringUtils.isNotBlank(value)) {

          try {
            //解析JSON
            val jSONObject = JSON.parseObject(value)

            val callId = jSONObject.getString("callId")
            if (StringUtils.isNotBlank(callId)) {

              out.collect(value)

            } else {
              //打印告警信息
              logger.warn(s"缺少 callId 字段: ${value}")
            }
          } catch {
            case e: Throwable => {
              logger.error(s"JSON解析异常: ${value}", e)
            }
          }

        }

      }
    }).keyBy(line => {
      val jSONObject = JSON.parseObject(line)

      jSONObject.getString("callId")
    })


    //将2个流进行连接
    val connecedDS: ConnectedStreams[String, String] = keyedCallLogDS.connect(keyedSatisfactionDS)

    val resultDS: DataStream[String] = connecedDS.flatMap(new RichCoFlatMapFunction[String, String, String] {
      /**
       * 记录通话记录的状态
       */
      private var callLogState: ValueState[String] = _

      /**
       * 记录满意度评价的状态
       */
      private var satisfactionState: ValueState[String] = _

      override def open(parameters: Configuration): Unit = {

        callLogState = getRuntimeContext.getState(
          new ValueStateDescriptor[String]("callLogState", classOf[String])
        )

        satisfactionState = getRuntimeContext.getState(
          new ValueStateDescriptor[String]("satisfactionState", classOf[String])
        )

      }

      override def flatMap1(value: String, out: Collector[String]): Unit = {
        //收到通话记录数据
        //先判断是否已经收到满意度评价消息
        val satisfactionJSONStr = satisfactionState.value()
        if (StringUtils.isNotEmpty(satisfactionJSONStr)) {
          //说明已经收到满意度评价了
          //开始拼接消息
          val callLogJSON = JSON.parseObject(value)
          val satisfactionJSON = JSON.parseObject(satisfactionJSONStr)

          //获取满意度评分
          val score = satisfactionJSON.getIntValue("score")
          // 清空满意度JSON对象
          satisfactionJSON.clear()

          //添加到通话记录中
          callLogJSON.put("score", score)

          //输出结果
          out.collect(callLogJSON.toJSONString)

          //通话记录JSON也清空一下
          callLogJSON.clear()

          //清空满意度的状态数据
          satisfactionState.clear()
        } else {
          //没有收到满意度消息
          //把收到的通话记录消息保存到状态中
          callLogState.update(value)
        }
      }

      override def flatMap2(value: String, out: Collector[String]): Unit = {

        //收到满意度评价数据
        //先判断是否已经收到通话记录消息
        val callLogJSONStr = callLogState.value()
        if (StringUtils.isNotEmpty(callLogJSONStr)) {
          //说明已经收到通话记录了
          //开始拼接消息
          val callLogJSON = JSON.parseObject(callLogJSONStr)
          val satisfactionJSON = JSON.parseObject(value)

          //获取满意度评分
          val score = satisfactionJSON.getIntValue("score")
          // 清空满意度JSON对象
          satisfactionJSON.clear()

          //添加到通话记录中
          callLogJSON.put("score", score)

          //输出结果
          out.collect(callLogJSON.toJSONString)

          //通话记录JSON也清空一下
          callLogJSON.clear()

          //把通话记录的状态清空
          callLogState.clear()
        } else {
          //没有收到通话记录消息
          //把收到的满意度消息保存到状态中
          satisfactionState.update(value)

        }

      }

      override def close(): Unit = {
        //程序退出不能清空状态，不然重启后状态数据就无法恢复了
      }
    })

    resultDS.print()

    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
